package cn.lagou.sparkcore.work

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import scala.math.{pow, sqrt}

/**
 * 假设：K=9
 * 将iris分成2个文件：
 * 文件A：包含141条数据。已知分类
 * 文件B：包含9条数据（每种类型各3条，把分类去掉）。未知分类
 * 算法：
 * 1.读文件A、B，形成数据集X、Y
 * 2.求数据集Y中每个点到数据集X中每个点的距离，得到数据集D
 * 3.找到数据集D中距离最小的K个点
 * 4.求K个点的分类情况
 * 5.距离使用欧几里得距离
 */
object Work4 {
  case class LabelPoint(label: String, point: Array[Double])
  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 业务逻辑
    // 1、标准设置为9
    val K = 9
    // 2、处理数据
    val lines = sc.textFile("./src/main/data/lris.csv")
      .map(line => {
        val fields = line.split(",")
        if (fields.length==5)
          LabelPoint("", fields.tail.map(_.toDouble))
        else
          LabelPoint(fields.last, fields.init.tail.map(_.toDouble))
      })
    // 3、将数据分为样本数据、测试数据
    val sampleRDD: RDD[LabelPoint] = lines.filter(_.label != "")
    val sampleData: Array[Array[Double]] = lines.filter(_.label == "").collect().map(_.point)

    // 计算距离
    def getDis(x: Array[Double], y: Array[Double]): Double = {
      sqrt(x.zip(y).map(z => pow(z._1 - z._2, 2)).sum)
    }

    // 4、求最近的K个点；对这K个点的label做wc，得到最终结果
    sampleData.foreach(elem => {
      // 获取所有距离
      val dists: RDD[(Double, String)] = sampleRDD.map(labelpoint => (getDis(elem, labelpoint.point), labelpoint.label))
      // 获取前九个
      val minDists: Array[(Double, String)] = dists.sortBy(_._1).take(K)
      val labels: Array[String] = minDists.map(_._2)
      print(s"${elem.toBuffer} : ")
      // 最终结果预测
      labels.groupBy(x=>x).mapValues(_.length).foreach(print)
      println()
    })
    // 关闭SparkContext
    sc.stop()
  }
}
